package com.geek.framework.study.event.distributed;

import com.geek.framework.study.event.EventListener;
import com.geek.framework.study.event.reactive.stream.ListenerSubscriberAdapter;
import com.geek.framework.study.reactive.quickstart.SimplePublisher;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPubSub;

import java.util.Date;
import java.util.EventObject;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @Author: ysq
 * @Date: 2021/9/1
 * @Description: 分布式事件_发布者
 */
public class DistributedEventPublisher {
  private final SimplePublisher<EventObject> simplePublisher;

  private final JedisPool jedisPool;

  private final ExecutorService executorService;

  public DistributedEventPublisher(String uri) {
    simplePublisher = new SimplePublisher();
    this.jedisPool = new JedisPool(uri);
    // Build-in listener
    addEventListener(event -> {
      if (event instanceof DistributedEventObject) {
        // Event -> Pub/Sub
        Jedis jedis = jedisPool.getResource();
        jedis.publish("test", (String) event.getSource());
        jedis.close();
      }
    });

    this.executorService = Executors.newSingleThreadExecutor();

    executorService.execute(() -> {
      Jedis jedis = jedisPool.getResource();
      jedis.subscribe(new JedisPubSub() {
        @Override
        public void onMessage(String channel, String message) {
          if ("test".equals(channel)) {
            publish(new EventObject(message));
          }
        }
      }, "test");
      jedis.close();
    });
  }

  // 发布_本地事件
  public void publish(Object event) {
    simplePublisher.publish(new DistributedEventObject(event));
  }

  // 发布_分布式事件
  private void publish(EventObject event) {
    simplePublisher.publish(event);
  }

  public void addEventListener(EventListener eventListener) {
    simplePublisher.subscribe(new ListenerSubscriberAdapter(eventListener));
  }

  public void close() {
    jedisPool.close();
    executorService.shutdown();
  }


  /**
   * 切换到redis目录下运行命令
   * Redis 运行
   * redis-server.exe redis.windows.conf
   *
   * 新建一个Terminal窗口
   * Redis 测试
   * redis-cli.exe -h 127.0.0.1 -p 6379
   */
  public static void main(String[] args) {
    DistributedEventPublisher eventPublisher = new DistributedEventPublisher("redis://127.0.0.1:6379");

    // Publish Event
    eventPublisher.publish(String.valueOf(new Date()));

    eventPublisher.close();
  }
}
